package engine

import (
	"errors"
	"gitee.com/moyrn/sa/store"
	"gitee.com/moyrn/sa/store/models"
	"net/http"
	"os"
	"time"
)

// 加入请求时间间隔控制
type Engine struct {
	Store          *store.Store     // 存储
	threadChan     chan models.Task // 任务队列
	threadNum      int              // 线程数量
	threadWaitTime []int            // 线程等待时间
	adapter        []Adapter        // 已注册的适配器
	HttpClient     *http.Client     // http客户端

	tick *time.Ticker // 请求间隔定时
}

/*
	初始化引擎
	@Params
	s			存储
	logPath 	日志路径
	tick 		定时器间隔时长(.eg: tick=number*time.Second)
	threadNum	线程数(默认1)
*/
func NewEngine(s *store.Store, logPath string, tick time.Duration, threadNum ...int) (*Engine, error) {
	logger, err := store.NewLogger(logPath)
	if err != nil {
		return nil, err
	}
	store.Log = logger
	if len(threadNum) == 0 {
		threadNum = []int{1}
	}
	var ticker *time.Ticker
	if tick != 0 {
		ticker = time.NewTicker(tick)
	}

	return &Engine{
		Store:          s,
		threadChan:     make(chan models.Task, 1),
		threadNum:      threadNum[0],
		threadWaitTime: make([]int, threadNum[0]),
		HttpClient:     &http.Client{Timeout: 10 * time.Second},
		tick:           ticker,
	}, nil
}

func (e *Engine) RegisterAdapters(adapters []Adapter) {
	for _, ad := range adapters {
		e.RegisterAdapter(ad)
		store.Log.Info(ad.ParserFunction, "register_success.")
	}
}

func (e *Engine) RegisterAdapter(adapter Adapter) {
	e.adapter = append(e.adapter, adapter)
}

// TODO 加入任务结束逻辑
func (e *Engine) Run() {
	// 定时检测，连续多次未发现新任务，直接结束
	for i := 0; i < e.threadNum; i++ {
		go func(gid int) {
			for {
				<-e.tick.C
				if err := e.do(gid); err != nil {
					store.Log.Info("[engine] do error", err)
					continue
				}
			}
		}(i)
	}

	var outCount int
	window := make([]error, 3)
	for {
		// 读取新任务
		line, err := e.Store.Read()
		if err == nil {
			// 读取成功，存入channel
			e.threadChan <- line
			continue
		}
		func() {
			defer time.Sleep(time.Millisecond * 200)
			// 移动划窗
			window[0], window[1], window[2] = window[1], window[2], err
			if window[0] == nil || window[1] == nil || window[2] == nil {
				// 三次未全部失败，延时重试
				return
			}
			// 所有协程均在等待状态，退出
			for _, v := range e.threadWaitTime {
				if v > 3 {
					outCount++
				}
			}
			if outCount == e.threadNum {
				store.Log.Info("engine stop")
				os.Exit(0)
			}
		}()
	}
}

func (e *Engine) do(gid int) error {
	defer func() {
		if err := recover(); err != nil {
			store.Log.Info("[engine] recover", err)
		}
	}()
	line, err := e.pull(gid)
	if err != nil {
		return err
	}

	adapter, err := e.findAdapter(line.ParserFunction)
	if err != nil {
		store.Log.Info("[line] error:", line.Url, line.ParserFunction, err)
		_ = e.push(line, err)
		return err
	}
	store.Log.Info("[line] detail", line.Url, line.ParserFunction)
	newLines, err := adapter.Adapter(e, line)
	if err := e.push(line, err); err != nil {
		return err
	}
	for _, v := range newLines {
		if err := e.SetNewTask(&v); err != nil {
			store.Log.Info("[engine] SetNewTask error", v, err)
		}
	}

	return nil
}

func (e *Engine) findAdapter(fn string) (Adapter, error) {
	for _, v := range e.adapter {
		if fn == v.ParserFunction {
			return v, nil
		}
	}

	return Adapter{}, errors.New("not_found_this_adapter")
}

func (e *Engine) pull(gid int) (*models.Task, error) {
	timer := time.NewTimer(time.Second * 2)
	defer timer.Stop()
	for {
		timer.Reset(time.Second * 2)
		select {
		case line := <-e.threadChan:
			return &line, nil
		case <-timer.C:
		}
		if e.threadWaitTime[gid] < 1000 {
			e.threadWaitTime[gid]++
		}
	}
}

func (e *Engine) SetNewTask(task *models.Task) error {
	return e.Store.Write(task)
}

func (e *Engine) push(task *models.Task, err ...error) error {
	if len(err) > 0 && err[0] != nil {
		return e.Store.SetLineFinished(task.ID, task.Title, task.Content, task.Remark, err[0].Error())
	}
	return e.Store.SetLineFinished(task.ID, task.Title, task.Content, task.Remark, "")
}

type Adapter struct {
	ParserFunction string
	Adapter        func(e *Engine, task *models.Task) (newTasks []models.Task, err error)
}
